This attempts to make code ES version agnostic between

7 and 8.

We import both python libraries `elasticsearch7` and `elasticsearch8`
We test the server version and set it in `settings.py`
We use the correct library based on the server major version

Author: smithellis

.env-test

@@ -22,4 +22,4 @@ REUSE_DB=0
 ENABLE_ADMIN=True
 SET_LOCALE_PATH=False
 SECURE_SSL_REDIRECT=False
-ES_URLS=http://elasticsearch:9200
+ES_URLS=http://elasticsearch:9200

docker-compose.yml

@@ -34,20 +34,18 @@ services:
       - pgvolume:/var/lib/postgresql/data
 
   elasticsearch:
-    image: docker.elastic.co/elasticsearch/elasticsearch:8.18.0
+    image: docker.elastic.co/elasticsearch/elasticsearch:7.10.2
     environment:
       - discovery.type=single-node
       - LOG4J_FORMAT_MSG_NO_LOOKUPS=true
-      - xpack.security.enabled=false
-      - xpack.security.http.ssl.enabled=false
     ports:
       - "9200:9200"
       - "9300:9300"
     volumes:
       - ./kitsune/search/dictionaries/synonyms:/usr/share/elasticsearch/config/synonyms
 
   kibana:
-    image: docker.elastic.co/kibana/kibana:8.18.0
+    image: docker.elastic.co/kibana/kibana:7.10.2
     ports:
       - 5601:5601
     environment:

kitsune/community/utils.py

@@ -6,8 +6,11 @@
 from django.core.cache import cache
 from django.db.models import Count, F, Q
 from django.db.models.functions import Now
-from elasticsearch.dsl import A
 
+if settings.ES_VERSION == 8:
+    from elasticsearch8.dsl import A
+else:
+    from elasticsearch_dsl import A
 from kitsune.products.models import Product
 from kitsune.search.documents import AnswerDocument, ProfileDocument
 from kitsune.users.models import ContributionAreas, User

kitsune/questions/models.py

@@ -21,7 +21,13 @@
 from django.urls import is_valid_path
 from django.utils import translation
 from django.utils.translation import pgettext
-from elasticsearch import ApiError
+
+if settings.ES_VERSION == 7:
+    from elasticsearch7 import ElasticsearchException
+    ESExceptionClass = ElasticsearchException
+else:
+    from elasticsearch8 import ApiError
+    ESExceptionClass = ApiError
 from product_details import product_details
 
 from kitsune.flagit.models import FlaggedObject
@@ -595,7 +601,7 @@ def related_documents(self):
                 for hit in search[:3].execute().hits
             ]
             cache.set(key, documents, settings.CACHE_LONG_TIMEOUT)
-        except ApiError:
+        except ESExceptionClass:
             log.exception("ES MLT related_documents")
             documents = []
 
@@ -641,7 +647,7 @@ def related_questions(self):
                 for hit in search[:3].execute().hits
             ]
             cache.set(key, questions, settings.CACHE_LONG_TIMEOUT)
-        except ApiError:
+        except ESExceptionClass:
             log.exception("ES MLT related_questions")
             questions = []
 

kitsune/search/base.py

@@ -10,11 +10,19 @@
 from django.utils import timezone
 from django.utils.translation import gettext as _
 from elasticsearch.exceptions import NotFoundError, RequestError
-from elasticsearch.dsl import Document as DSLDocument
-from elasticsearch.dsl import InnerDoc, MetaField
-from elasticsearch.dsl import Search as DSLSearch
-from elasticsearch.dsl import field
-from elasticsearch.dsl.utils import AttrDict
+
+if settings.ES_VERSION == 8:
+    from elasticsearch8.dsl import Document as DSLDocument
+    from elasticsearch8.dsl import InnerDoc, MetaField
+    from elasticsearch8.dsl import Search as DSLSearch
+    from elasticsearch8.dsl import field
+    from elasticsearch8.dsl.utils import AttrDict
+else:
+    from elasticsearch_dsl import Document as DSLDocument
+    from elasticsearch_dsl import InnerDoc, MetaField
+    from elasticsearch_dsl import Search as DSLSearch
+    from elasticsearch_dsl import field
+    from elasticsearch_dsl.utils import AttrDict
 from pyparsing import ParseException
 
 from kitsune.search.config import (
@@ -205,7 +213,10 @@ def to_action(self, action=None, is_bulk=False, **kwargs):
         # documents will be updated/added directly in the index.
         # For ES8 we need to use the string value "true" instead of boolean
         if settings.TEST and not is_bulk:
-            kwargs.update({"refresh": "true"})
+            if settings.ES_VERSION >= 8:
+                kwargs.update({"refresh": "true"})
+            else:
+                kwargs.update({"refresh": True})
 
         if not action or action == "index":
             return payload if is_bulk else self.save(**kwargs)
@@ -320,7 +331,8 @@ class inherits, relevant to the documents the child class is searching over.
     """
 
     total: int = dfield(default=0, init=False)
-    hits: AttrDict = dfield(default_factory=list, init=False)
+    # Use a single declaration with a type that works for both ES versions
+    hits: Union[list[AttrDict], AttrDict] = dfield(default_factory=list, init=False)
     results: list[dict] = dfield(default_factory=list, init=False)
     last_key: Union[int, slice, None] = dfield(default=None, init=False)
 
@@ -394,14 +406,18 @@ def run(self, key: Union[int, slice] = slice(0, settings.SEARCH_RESULTS_PER_PAGE
                 return self.run(key)
             raise e
 
-        self.hits = cast(AttrDict, result.hits)
+        if settings.ES_VERSION >= 8:
+            self.hits = cast(AttrDict, result.hits)
+        else:
+            self.hits = result.hits
         self.last_key = key
 
         # Handle total hits according to ES8 response format
         # In ES8, total is always returned as an object with a 'value' property
-        self.total = getattr(self.hits.total, "value", 0)
-        if isinstance(self.hits.total, dict):
-            self.total = self.hits.total.get("value", 0)
+        if settings.ES_VERSION >= 8:
+            self.total = getattr(self.hits.total, "value", 0)
+            if isinstance(self.hits.total, dict):
+                self.total = self.hits.total.get("value", 0)
 
         self.results = [self.make_result(hit) for hit in self.hits]
 

kitsune/search/documents.py

@@ -1,5 +1,10 @@
 from django.db.models import Count, Prefetch, Q
-from elasticsearch.dsl import InnerDoc, connections, field
+from django.conf import settings
+
+if settings.ES_VERSION == 8:
+    from elasticsearch8.dsl import InnerDoc, connections, field
+else:
+    from elasticsearch_dsl import InnerDoc, connections, field
 
 from kitsune.forums.models import Post
 from kitsune.questions.models import Answer, Question

kitsune/search/es_utils.py

@@ -6,7 +6,11 @@
 from elasticsearch import Elasticsearch
 from elasticsearch.helpers import bulk as es_bulk
 from elasticsearch.helpers.errors import BulkIndexError
-from elasticsearch.dsl import Document, UpdateByQuery, analyzer, char_filter, token_filter
+
+if settings.ES_VERSION == 8:
+    from elasticsearch8.dsl import Document, UpdateByQuery, analyzer, char_filter, token_filter
+else:
+    from elasticsearch_dsl import Document, UpdateByQuery, analyzer, char_filter, token_filter
 
 from kitsune.search import config
 
@@ -46,6 +50,7 @@ def _create_synonym_graph_filter(synonym_file_name):
         synonyms_path=f"synonyms/{synonym_file_name}.txt",
         expand="true",
         lenient="true",
+        updateable="true",
     )
 
 
@@ -94,16 +99,23 @@ def es_client(**kwargs):
     if es_cloud_id := settings.ES_CLOUD_ID:
         kwargs.update({"cloud_id": es_cloud_id, "basic_auth": settings.ES_HTTP_AUTH})
     else:
-        # Elasticsearch 8.x settings
+        # Basic ES settings that apply to all versions
         es_settings = {
             "hosts": settings.ES_URLS,
-            "request_timeout": settings.ES_TIMEOUT,
-            "retry_on_timeout": settings.ES_RETRY_ON_TIMEOUT,
-            # SSL settings - these are needed for ES8 which requires SSL by default
-            "verify_certs": settings.ES_VERIFY_CERTS,
-            "ssl_show_warn": settings.ES_SSL_SHOW_WARN,
         }
 
+        # Add settings that are specific to ES 8+
+        if getattr(settings, "ES_VERSION", 0) >= 8:
+            es_settings.update(
+                {
+                    "request_timeout": settings.ES_TIMEOUT,
+                    "retry_on_timeout": settings.ES_RETRY_ON_TIMEOUT,
+                    # SSL settings - these are needed for ES8 which requires SSL by default
+                    "verify_certs": settings.ES_VERIFY_CERTS,
+                    "ssl_show_warn": settings.ES_SSL_SHOW_WARN,
+                }
+            )
+
         if settings.ES_HTTP_AUTH:
             es_settings.update({"basic_auth": settings.ES_HTTP_AUTH})
 
@@ -149,7 +161,10 @@ def index_object(doc_type_name, obj_id):
     kwargs = {}
     # For ES8, use string "true" instead of boolean True for refresh parameter
     if settings.TEST:
-        kwargs["refresh"] = "true"
+        if settings.ES_VERSION >= 8:
+            kwargs["refresh"] = "true"
+        else:
+            kwargs["refresh"] = True
 
     if doc_type.update_document:
         doc_type.prepare(obj).to_action("update", doc_as_upsert=True, **kwargs)
@@ -191,7 +206,11 @@ def index_objects_bulk(
         (doc.to_action(action=action, is_bulk=True, **kwargs) for doc in docs),
         chunk_size=elastic_chunk_size,
         raise_on_error=False,  # we'll raise the errors ourselves, so all the chunks get sent
-        refresh="true" if settings.TEST else False,  # use string "true" for ES8 compatibility
+        refresh=(
+            "true"
+            if settings.TEST and settings.ES_VERSION >= 8
+            else (True if settings.TEST else False)
+        ),  # refresh parameter based on test mode and ES version
     )
     errors = [
         error
@@ -212,17 +231,39 @@ def remove_from_field(doc_type_name, field_name, field_value):
     doc_type = next(cls for cls in get_doc_types() if cls.__name__ == doc_type_name)
 
     # Create script as a string
-    script_source = (
-        f"if (ctx._source.{field_name} != null) {{ "
-        f"ctx._source.{field_name}.removeAll(Collections.singleton(params.value)); "
-        f"}}"
-    )
+    if getattr(settings, "ES_VERSION", 0) >= 8:
+        script_source = (
+            f"if (ctx._source.{field_name} != null) {{ "
+            f"ctx._source.{field_name}.removeAll(Collections.singleton(params.value)); "
+            f"}}"
+        )
+    else:
+        script_source = (
+            f"if (ctx._source.{field_name}.contains(params.value)) {{"
+            f"ctx._source.{field_name}.remove(ctx._source.{field_name}.indexOf(params.value))"
+            f"}}"
+        )
 
     # Set up the update query
     update = UpdateByQuery(using=es_client(), index=doc_type._index._name)
 
     # Apply the script with parameters
-    update = update.script(source=script_source, lang="painless", params={"value": field_value})
+    if getattr(settings, "ES_VERSION", 0) >= 8:
+        update = update.script(
+            source=script_source, lang="painless", params={"value": field_value}
+        )
+    else:
+        # For ES7 and below, we need to filter documents explicitly
+        update = update.filter("term", **{field_name: field_value})
+        update = update.script(
+            source=script_source,
+            lang="painless",
+            params={"value": field_value},
+            conflicts="proceed",
+        )
+
+    # refresh index to ensure search fetches all matches
+    doc_type._index.refresh()
 
     update.execute()
 
@@ -238,6 +279,9 @@ def delete_object(doc_type_name, obj_id):
     kwargs = {}
     # For ES8, use string "true" instead of boolean True for refresh parameter
     if settings.TEST:
-        kwargs["refresh"] = "true"
+        if settings.ES_VERSION >= 8:
+            kwargs["refresh"] = "true"
+        else:
+            kwargs["refresh"] = True
 
     doc.to_action("delete", **kwargs)

kitsune/search/fields.py

@@ -1,9 +1,15 @@
 from functools import partial
 
 from django.conf import settings
-from elasticsearch.dsl.field import Keyword
-from elasticsearch.dsl.field import Object as DSLObject
-from elasticsearch.dsl.field import Text
+
+if settings.ES_VERSION == 8:
+    from elasticsearch8.dsl.field import Keyword
+    from elasticsearch8.dsl.field import Object as DSLObject
+    from elasticsearch8.dsl.field import Text
+else:
+    from elasticsearch_dsl.field import Keyword
+    from elasticsearch_dsl.field import Object as DSLObject
+    from elasticsearch_dsl.field import Text
 
 from kitsune.search.es_utils import es_analyzer_for_locale
 

kitsune/search/management/commands/es_init.py

@@ -1,5 +1,10 @@
 from django.core.management.base import BaseCommand
-from elasticsearch.dsl.exceptions import IllegalOperation
+from django.conf import settings
+
+if settings.ES_VERSION == 8:
+    from elasticsearch8.dsl.exceptions import IllegalOperation
+else:
+    from elasticsearch_dsl.exceptions import IllegalOperation
 from datetime import datetime, timezone
 
 from kitsune.search.es_utils import get_doc_types, es_client
@@ -76,6 +81,9 @@ def handle(self, *args, **kwargs):
             index = dt.alias_points_at(dt.Index.write_alias)
             if kwargs["reload_search_analyzers"]:
                 print(f"Reloading search analyzers on {index}")
-                client.indices.reload_search_analyzers(index=[index])
+                if settings.ES_VERSION >= 8:
+                    client.indices.reload_search_analyzers(index=[index])
+                else:
+                    client.indices.reload_search_analyzer(index)
 
             print("")  # print blank line to make console output easier to read

kitsune/search/parser/operators.py

@@ -1,4 +1,9 @@
-from elasticsearch.dsl import Q as DSLQ
+from django.conf import settings
+
+if settings.ES_VERSION == 8:
+    from elasticsearch8.dsl import Q as DSLQ
+else:
+    from elasticsearch_dsl import Q as DSLQ
 
 from .tokens import BaseToken
 

kitsune/search/parser/tokens.py

@@ -1,6 +1,10 @@
 from abc import ABC, abstractmethod
+from django.conf import settings
 
-from elasticsearch.dsl import Q as DSLQ
+if settings.ES_VERSION == 8:
+    from elasticsearch8.dsl import Q as DSLQ
+else:
+    from elasticsearch_dsl import Q as DSLQ
 
 
 class BaseToken(ABC):

kitsune/search/search.py

@@ -5,8 +5,13 @@
 
 import bleach
 from dateutil import parser
+from django.conf import settings
 from django.utils.text import slugify
-from elasticsearch.dsl import Q as DSLQ
+
+if settings.ES_VERSION == 8:
+    from elasticsearch8.dsl import Q as DSLQ
+else:
+    from elasticsearch_dsl import Q as DSLQ
 
 from kitsune.products.models import Product
 from kitsune.search import HIGHLIGHT_TAG, SNIPPET_LENGTH